package com.hadoop.fof;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

import java.io.IOException;

public class FMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    Text mkey = new Text();
    IntWritable mvalue = new IntWritable();

    protected void map(LongWritable key, Text value,
                       Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //输入一行数据，迭代后两两输出，输出时标注关系， 直接好友:0，间接好友:1
        String [] friends = StringUtils.split(value.toString(),' ');
        for (int i = 0 ;i < friends.length ; i++){
            mkey.set(compare(friends[0],friends[i]));
            mvalue.set(0);

            context.write(mkey,mvalue);

            for(int j = i +1;j<friends.length-1;j++){
                mkey.set(compare(friends[i+1],friends[j+1]));
                mvalue.set(1);
                /*
                 * Cat-Hadoop 1
                 * Cat-Hello 1
                 * Hadoop-Hello 1
                 */
                context.write(mkey,mvalue);

            }
        }
    }

    public static String compare(String s1 , String s2){
        if(s1.compareTo(s2) > 0){
            return s2 + "-" + s1 ;
        }else{
            return s1 + "-" + s2 ;
        }
    }


}
